package com.etlmaster.rowprocessor.core;

import java.util.Map;

import com.etlmaster.executor.trans.core.Line;
import com.etlmaster.executor.trans.core.LineMeta;
import com.etlmaster.executor.utils.LogWriter;
import com.etlmaster.rowprocessor.disruptor.readEvent.LineEvent;


/**
 * 数据源步骤，行级处理的开输入
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
public abstract class AbstractReader  extends AbstractOutPutPlugin implements Runnable{
	private String outPutStreamId;
	public static final int 
	NOT_STARTED = -10, //未启动
	ABORTED = -1, //异常中止
	FINISHED=1,
	RUNNING = 10, //运行中
	SUSPENDED = 11; //挂起

	private int status=NOT_STARTED;
	public AbstractReader(String id,String jobId, String name, String code,
			Map<String, Object> paramMap,Map<String, Object> var) {
		super(id,jobId, name, code, paramMap,var);
	}

	public void run(){
		setStatus(RUNNING);
		try{
			write();
		}catch(Exception e){
			e.printStackTrace();
			LogWriter.addLog(this.getJobId(),0,"ERROR",e.getMessage());
		}
		setStatus(FINISHED);
	}
	public void updateMeta(String sourceTab,String sourceCol,LineMeta meta){
		if(!meta.getColIndex().containsKey(sourceTab.toUpperCase()+"_"+sourceCol.toUpperCase())){
			meta.addField(sourceTab, sourceCol);
		}
	}
    public void sendData(Line line)
    {
        long sequence = getRingBuffer().next();  // Grab the next sequence
        try
        {
        	LineEvent event = getRingBuffer().get(sequence); // Get the entry in the Disruptor
                                                        // for the sequence
            event.set(line);  // Fill with data
        }
        finally
        {
            getRingBuffer().publish(sequence);
        }
        this.setCounter(sequence);
        LogWriter.addLog(this.getJobId(),0,"debug","reader sequence:"+sequence);
    }


	public String[] getOutPutStreamId() {
		return outPutStreamId.split(",");
	}

	public void setOutPutStreamId(String outPutStreamId) {
		this.outPutStreamId = outPutStreamId;
	}

	public int getStatus() {
		return status;
	}

	public void setStatus(int status) {
		this.status = status;
	}

}
